package com.example.kafka.service;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @author wkj
 * Kafka 主题管理服务类
 */
@Service
public class KafkaTopicService {

    @Autowired
    private AdminClient adminClient;

    /**
     * 创建主题
     */
    public void createTopic(String name, int partitions, short replicationFactor) throws Exception {
        NewTopic newTopic = new NewTopic(name, partitions, replicationFactor);
        CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic));
        result.all().get(); // 阻塞直到完成
    }

    /**
     * 删除主题
     */
    public void deleteTopic(String name) throws Exception {
        DeleteTopicsResult result = adminClient.deleteTopics(Collections.singleton(name));
        result.all().get();
    }

    /**
     * 获取所有主题名称
     */
    public Set<String> listTopics() throws Exception {
        ListTopicsResult topics = adminClient.listTopics();
        return topics.names().get();
    }

    /**
     * 查看某个主题的详细描述（分区、副本）
     */
    public Map<String, TopicDescription> describeTopic(String name) throws Exception {
        DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(name));
        // 推荐方式（支持 Kafka 3.x+）
        Map<String, KafkaFuture<TopicDescription>> topicFutures = result.topicNameValues();
        Map<String, TopicDescription> descriptions = new HashMap<>();

        for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : topicFutures.entrySet()) {
            descriptions.put(entry.getKey(), entry.getValue().get());
        }

        return descriptions;
    }
}
